package cn.edu.gdut.deepcode.component;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * <p>
 *     redis的stream流服务
 * </p>
 *
 * @Author: 暮倦
 * @Since: 2025-04-15  19:24
 * @Version: 1.0.0
 */

@Slf4j
@Component
@Getter
public class RedisStreamService {
    private final StringRedisTemplate redisTemplate;

    private final StreamOperations<String, Object, Object> opsForStream;

    public RedisStreamService(StringRedisTemplate redisTemplate){
        this.redisTemplate=redisTemplate;
        this.opsForStream=redisTemplate.opsForStream();
    }
    /*
    新增数据到stream流
     */
    public void addDataToStream(String streamKey, Map<String,String> data){
        opsForStream.add(streamKey,data);
    }


    /*
    创建消费者组
     */
    public void addGroup(String streamName,String groupName){
        opsForStream.createGroup(streamName,groupName);
    }

    /*
    删除消费者组
     */
    public void removeGroup(String streamName,String groupName){
        opsForStream.destroyGroup(streamName,groupName);
    }

    /*
    从stream流中拉取数据(强制优先拉取pending中的数据)
     */
    public <T> List<Map<Object, Object>> pollDataFromStream(String streamName, String groupName, String consumerName, int batchSize){
        // 获取消费者
        Consumer consumer = Consumer.from(groupName, consumerName);
        // 生成拉取配置
        StreamReadOptions options = StreamReadOptions.empty()
                .count(batchSize)
                .block(Duration.ofSeconds(5L));// 5秒后没数据就会取消拉取数据

        // 设置读取位置（从尚未消费的消息开始,强制优先拉取pending中的数据）
        StreamOffset<String> offset = StreamOffset.create(streamName, ReadOffset.from("0-0"));

        List<MapRecord<String, Object, Object>> records = opsForStream.read(consumer, options, offset);
        if (records==null){
            records= Collections.emptyList();
        }
        return records.stream().map(Record::getValue).toList();
    }
    /*
    批量确认已消费数据
     */
    public void batchAckData(String streamName,String groupName,String... ids){
            opsForStream.acknowledge(streamName,groupName,ids);
    }

}


